package com.github.kezhenxu94.spark;

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;

/**
 * Created by kezhenxu on 1/2/17.
 */
public class TestSpark {

    public static void main(String[] args) throws InterruptedException {
        final SparkConf sparkConf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount");
        final JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, Durations.seconds(1));
        final JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost", 9999);
        final JavaDStream<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
        words.mapToPair(word -> new Tuple2<>(word, 1))
                .reduceByKey(($1, $2) -> $1 + $2)
                .print();
        jssc.start();
        jssc.awaitTermination();
    }
}
